package com.atguigu.edu.app.dwd.db;

import com.atguigu.edu.app.BaseSQLApp;
import com.atguigu.edu.common.Constant;
import com.atguigu.edu.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


/*
* 作废，放至DwdBaseDb
* */
public class Dwd_02_DwdCartAddApp extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_02_DwdCartAddApp().init(
                3002,2,
                "Dwd_02_DwdCartAddApp"
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 1. 读取 ods_db 数据
        readOdsDb(tEnv, "Dwd_02_DwdCartAddApp");

        // 2. 过滤出加购表数据
        Table cartInfo = tEnv.sqlQuery("select " +
                "`data`['id'] id," +
                "`data`['user_id'] user_id," +
                "`data`['course_id'] course_id," +
                "ts " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='cart_info' " +
                "and `type`='insert' ");
//        cartInfo.execute().print();
        tEnv.createTemporaryView("cart_info", cartInfo);

        // 3.写出到kafka
        tEnv.executeSql("create table dwd_trade_cart_add(" +
                "id string," +
                "user_id string," +
                "course_id string," +
                "ts bigint" +
                ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_CART_ADD));

        cartInfo.executeInsert("dwd_trade_cart_add");
    }
}
